package com.example.binlog.client;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSONObject;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;

import java.io.Serializable;
import java.util.*;

/**
 * @author yousj
 * @date 2021/4/13
 */
public class BinlogClient {

    // 监听的数据库
    private static final String HOST = "127.0.0.1";
    private static final int PORT = 3306;
    private static final String SCHEMA = "book";
    private static final String USERNAME = "root";
    private static final String PASSWORD = "root";
    // 所有表
    private static final Map<Long, String> TABLE_MAP = new HashMap<>();
    // 需要监听的表
    public static List<String> DATABASE_LIST = Arrays.asList("tb_01", "tb_02", "tb_03");
    // 监听的表的所有列名称
    // 根据表名称查询出所有列名称SQL语句, 封装到map中
    // COLUMNS_MAP: key => 表名, value => 表的所有列名
    // select mc.COLUMN_NAME from information_schema.COLUMNS mc where TABLE_SCHEMA = (select database()) and TABLE_NAME=#{tableName}
    public static Map<String, List<String>> COLUMNS_MAP = new HashMap<>();

    public static void main(String[] args) {

        BinaryLogClient client = new BinaryLogClient(HOST, PORT, SCHEMA, USERNAME, PASSWORD);
        // 监听服务id, 唯一
        client.setServerId(1);
        // 设置起始读取文件名称
        client.setBinlogFilename("");
        // 设置起始读取位置
        // client.setBinlogPosition(0L);

        // 注册监听
        client.registerEventListener((event -> {
            // 处理事件
            event(event, client);
        }));
    }

    private static void event(Event event, BinaryLogClient client) {
        // 获取监听结果数据集
        EventData data = event.getData();
        if (data != null) {
            if (data instanceof TableMapEventData) {
                TableMapEventData tableMapEventData = (TableMapEventData) data;
                TABLE_MAP.put(tableMapEventData.getTableId(), tableMapEventData.getTable());
            }
            // 新增
            if (data instanceof WriteRowsEventData) {
                WriteRowsEventData writeRowsEventData = (WriteRowsEventData) data;
                String tableName = TABLE_MAP.get(writeRowsEventData.getTableId());
                if (tableName != null && DATABASE_LIST.contains(tableName)) {
                    for (Serializable[] row : writeRowsEventData.getRows()) {
                        List<Object> objectList = (List<Object>) JSONObject.toJSON(row);
                    }
                }
            }
            // 删除
            else if (data instanceof DeleteRowsEventData) {
                DeleteRowsEventData deleteRowsEventData = (DeleteRowsEventData) data;
                String tableName = TABLE_MAP.get(deleteRowsEventData.getTableId());
                if (tableName != null && DATABASE_LIST.contains(tableName)) {
                    for (Serializable[] row : deleteRowsEventData.getRows()) {
                        List<Object> objectList = (List<Object>) JSONObject.toJSON(row);
                    }
                }
            }
            // 修改
            else if (data instanceof UpdateRowsEventData) {
                UpdateRowsEventData updateRowsEventData = (UpdateRowsEventData) data;
                String tableName = TABLE_MAP.get(updateRowsEventData.getTableId());
                if (tableName != null && DATABASE_LIST.contains(tableName)) {
                    for (Map.Entry<Serializable[], Serializable[]> row : updateRowsEventData.getRows()) {
                        // 修改之前的数据
                        List<Object> beforeObjectList = (List<Object>) JSONObject.toJSON(row.getKey());
                        // 修改之后的数据
                        List<Object> afterObjectList = (List<Object>) JSONObject.toJSON(row.getValue());
                    }
                }
            }
        }
    }

    /**
     * binlog监听操作记录值转化成对象
     *
     * @param columnNames 列名
     * @param values      值
     * @param beanClass   对象class
     * @return <T>
     */
    private <T> T toBean(List<String> columnNames, List<Object> values, Class<T> beanClass) {
        Map<String, Object> map = new HashMap<>();
        int len = columnNames.size();
        for (int i = 0; i < len; i++) {
            map.put(columnNames.get(i), values.get(i));
        }
        // map->bean 忽略大小写
        return BeanUtil.mapToBeanIgnoreCase(map, beanClass, false);
    }

}
